package ml

import base.BaseModel
import org.apache.spark.ml.clustering.{KMeans, KMeansModel}
import org.apache.spark.ml.feature.VectorAssembler
import org.apache.spark.sql._

import scala.collection.immutable

/*
此代码用于计算用户的价格敏感度
 */
object PSMModel  extends  BaseModel{
  //程序名
  override def setAppName: String = "PSMModel"
  //价格敏感度TagID
  override def setFourId: String = "151"


  //计算新标签
  override def getNewTag(spark: SparkSession, fiveTagDF: DataFrame, hbaseDF: DataFrame): DataFrame = {
    //fiveTagDF.show()
    //hbaseDF.show()
/*
+----+---+
|rule| id|
+----+---+
|   0|178|
|   1|179|
|   2|180|
|   3|181|
|   4|182|
+----+---+
+---------+-------------------+-----------+---------------+
| memberId|            orderSn|orderAmount|couponCodeValue|
+---------+-------------------+-----------+---------------+
| 13823431| ts_792756751164275|    2479.45|           0.00|
|  4035167| D14090106121770839|    2449.00|           0.00|
|  4035291| D14090112394810659|    1099.42|           0.00|
|  4035041| fx_787749561729045|    1999.00|           0.00|
| 13823285| D14092120154435903|    2488.00|           0.01|
|  4034219| D14092120155620305|    3449.00|         100.00|
|138230939|top_810791455519102|    1649.00|           0.00|
 */

    //引入隐式转换
    import spark.implicits._
    import org.apache.spark.sql.functions._


    //>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>第1步 计算PSM<<<<<<<<<<<<<<<<<<<
    /*11111111111111111
    优惠次数
    总购买次数
    优惠总金额
    应收总金额
     */

    //    优惠次数?????
    //couponCodeValue   等于0.00  表示没有优惠，没有优惠返回0
    //couponCodeValue 不等于0.00  表示有优惠  ，  有优惠返回1  ，使用sum(字段)  =>优惠次数
    var couponCodeState=functions
      .when('couponCodeValue===0.0D,0)
      .when('couponCodeValue=!=0.0D,1)
      .as("state")

    //一个商品200元，优惠30元，成交价170元   应收是200
    //应收总金额=成交价+优惠的金额
    var amountReceivable=col("orderAmount")+col("couponCodeValue") as ("amountReceivable")
    val tmpDF: DataFrame = hbaseDF.select('memberId,'orderSn,'orderAmount,'couponCodeValue,amountReceivable,couponCodeState)
/*
+---------+-------------------+-----------+---------------+----------------+-----+
| memberId|            orderSn|orderAmount|couponCodeValue|amountReceivable|state|
+---------+-------------------+-----------+---------------+----------------+-----+
| 13822859| ts_792756751164275|    2479.45|           0.00|         2479.45|    0|
| 13822859| D14092120165516725|    1999.00|           0.00|          1999.0|    0|
| 13822859| D14092120165617127|    1899.00|           0.00|          1899.0|    0|

| 13823431| D14092120170327337|    1699.00|         200.00|          1899.0|    1|
| 13823431|       RRS_15244971|    1558.00|           0.00|          1558.0|    0|
| 13823431| D14092120190458957|    1699.00|         200.00|          1899.0|    1|

| 13823281| D14092120194091167|     679.00|           0.00|           679.0|    0|
+---------+-------------------+-----------+---------------+----------------+-----+
 */


    //优惠次数
    var getCouponCodeSum=functions.sum('state) as ("couponCodeSum")
    //总购买次数
    var getAllSum=functions.count('state) as ("allSum")
    //优惠总金额
    var getCouponCodeAllMoney=functions.sum('couponCodeValue) as ("couponCodeAllMoney")
    //应收总金额
    var getReceivableAllMoney=functions.sum('amountReceivable) as ("receivableAllMoney")


    /*222222222222222222222222
优惠订单占比=   优惠次数 /总购买次数

平均优惠金额占比
    平均优惠金额/平均应收金额
        平均优惠金额=优惠总金额/优惠总次数
        平均应收金额=应收总金额 /订单总次数

优惠总金额占比=优惠总金额/应收总金额
 */
   //优惠订单占比
   var getCouponCodePercent   =col("couponCodeSum")/col("allSum")
    //平均优惠金额占比
   var getAvgCouponCodeMoneyPercent=
     (col("couponCodeAllMoney")/col("couponCodeSum"))/(col("receivableAllMoney")/col("allSum"))
    //优惠总金额占比
    var getCouponCodeAllMoneyPercent=col("couponCodeAllMoney")/col("receivableAllMoney")


    //33333333333333333
    //psm = 优惠订单占比 + 平均优惠金额占比 + 优惠总金额占比
    var getPSM=getCouponCodePercent+getAvgCouponCodeMoneyPercent+getCouponCodeAllMoneyPercent as ("PSM")

    val PSMDF: Dataset[Row] = tmpDF.groupBy('memberId)
      .agg(getCouponCodeSum, getAllSum, getCouponCodeAllMoney, getReceivableAllMoney)
      .select('memberId, getPSM)
      .filter('PSM isNotNull)

/*
+---------+-------------+------+------------------+------------------+
|memberId |couponCodeSum|allSum|couponCodeAllMoney|receivableAllMoney|
+---------+-------------+------+------------------+------------------+
|4033473  |3            |142   |500.0             |252430.92         |
|13822725 |4            |116   |800.0             |180098.34         |
|13823681 |1            |108   |200.0             |169946.1          |
|138230919|3            |125   |600.0             |240661.56999999998|

 */

/*//当有错误数据时，spark返回null
KMeans算法不能有null
+---------+--------------------+
|memberId |   PSM              |
+---------+--------------------+
|4033473  |0.11686252330855691 |
|13822725 |0.16774328728519597 |
|13823681 |0.13753522440350205 |
|138230919|0.1303734438365045  |
|13823083 |0.1380506927739941  |
|13823431 |0.15321482374431458 |
|4034923  |0.13927276336831218 |
|4033575  |0.11392752155030905 |
|13823153 |0.15547466292943982 |
|4034191  |0.11026694172505715 |
|4033483  |0.15976480445774954 |
|4033348  |0.13600092999496663 |
|4034761  |0.13114429909634118
 */
    //>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>第2步 数据向量化<<<<<<<<<<<<<<<<<<<
    //特征单词
    val featureStr = "feature"
    //最终的预测单词
    val predictStr = "predict"


    val PSMFeature: DataFrame = new VectorAssembler()
      .setInputCols(Array("PSM"))
      .setOutputCol(featureStr)
      .transform(PSMDF)

/*
+---------+-------------------+---------------------+
|memberId |PSM                |feature              |
+---------+-------------------+---------------------+
|4033473  |0.11686252330855691|[0.11686252330855691]|
|13822725 |0.16774328728519597|[0.16774328728519597]|
|13823681 |0.13753522440350205|[0.13753522440350205]|
|138230919|0.1303734438365045 |[0.1303734438365045] |
|13823083 |0.1380506927739941 |[0.1380506927739941] |
|13823431 |0.15321482374431458|[0.15321482374431458]|
 */

    //>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>第3步 使用KMeans分类<<<<<<<<<<<<<<<<<<<

    val model: KMeansModel = new KMeans()
      .setK(5)
      .setMaxIter(5)
      .setSeed(10)
      .setFeaturesCol(featureStr)
      .setPredictionCol(predictStr)
      .fit(PSMFeature)


    val PSMPredictDF: DataFrame=model.transform(PSMFeature)
    PSMPredictDF.show()
/*
+---------+-------------------+--------------------+-------+
| memberId|                PSM|             feature|predict|
+---------+-------------------+--------------------+-------+
|  4034761|0.13114429909634118|[0.13114429909634...|      0|
|  4035131|0.39147259141650464|[0.39147259141650...|      2|
|        7|0.13128367494800738|[0.13128367494800...|      0|
|138230911|0.11824736419948352|[0.11824736419948...|      0|
|  4035219|0.08062873693483025|[0.08062873693483...|      3|
+---------+-------------------+--------------------+-------+
 */

    //>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>第4步 从高到底将分类排序<<<<<<<<<<<<<<<<<<<

    val clusterCentersSum: immutable.IndexedSeq[(Int, Double)] = for(a<-model.clusterCenters.indices) yield (a,model.clusterCenters(a).toArray.sum)
    val clusterCentersSumSort: immutable.IndexedSeq[(Int, Double)] = clusterCentersSum.sortBy(_._2)

    //>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>第5步 获取排序后分区数据个索引<<<<<<<<<<<<<<<<<<<
    val clusterCenterIndex: immutable.IndexedSeq[(Int, Int)] = for(b<-clusterCentersSumSort.indices) yield (clusterCentersSumSort(b)._1,b)
    //  clusterCenterIndex.foreach(println)
  /*
  (3,0)
  (0,1)
  (4,2)
  (2,3)
  (1,4)
   */
    /*
+----+---+
|rule| id|
+----+---+
|   0|178|
|   1|179|
|   2|180|
|   3|181|
|   4|182|
+----+---+
     */

      //>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>第6步 与五级标签数据join，  计算分类ID个TagID<<<<<<<<<<<<<<<<<<<
      val clusterCenterIndexDF: DataFrame = clusterCenterIndex.toDF(predictStr,"Index")
      val predictTagID: DataFrame = clusterCenterIndexDF.join(fiveTagDF, clusterCenterIndexDF.col("Index") === fiveTagDF.col("rule"))
      .select(predictStr, "id")
    predictTagID.show()
    /*
+-------+---+
|predict| id|
+-------+---+
|      3|178|
|      0|179|
|      4|180|
|      2|181|
|      1|182|
+-------+---+
     */
    /*
+---------+-------------------+--------------------+-------+
| memberId|                PSM|             feature|predict|
+---------+-------------------+--------------------+-------+
|  4034761|0.13114429909634118|[0.13114429909634...|      0|
|  4035131|0.39147259141650464|[0.39147259141650...|      2|
|        7|0.13128367494800738|[0.13128367494800...|      0|
|138230911|0.11824736419948352|[0.11824736419948...|      0|
|  4035219|0.08062873693483025|[0.08062873693483...|      3|
+---------+-------------------+--------------------+-------+
     */
    //>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>第7步 join  分类后的数据jion得到用户ID和TagId<<<<<<<<<<<<<<<<<<<
/*
    val newTags: DataFrame =PSMPredictDF.join(predictTagID, PSMPredictDF.col("predict") === predictTagID.col("predict"))
      .select('memberId, 'id)
*/


    val predictTag: Map[String, String] = predictTagID.as[(String,String)].collect().toMap
    var getTagIs =udf((predict:String )=>{
      predictTag.get(predict)
    })
    val newTags: DataFrame = PSMPredictDF.select('memberId.as("userId"),getTagIs('predict).as("tagsId"))
    //newTags.show()
/*
+---------+------+
|   userId|tagsId|
+---------+------+
|  4033473|   179|
| 13822725|   179|
| 13823681|   179|
|138230919|   179|
| 13823083|   179|
| 13823431|   179|
|  4034923|   179|

 */

    //>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>第8步 返回结果<<<<<<<<<<<<<<<<<<<
    //newTags.show( false)
    newTags
  }

  //程序的入口
  def main(args: Array[String]): Unit = {
    exec()
  }


}

